package com.lhd.app.processor;

import com.lhd.app.bean.PetAgeLabel;
import com.lhd.app.bean.UserBehavior;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.util.*;

public class PetAgeLabelProcessor extends ProcessWindowFunction<UserBehavior, PetAgeLabel, String, TimeWindow> {
    
    @Override
    public void process(String userId, Context context, 
                      Iterable<UserBehavior> behaviors, 
                      Collector<PetAgeLabel> out) {
        
        Map<String, Map<String, Double>> petAgeScores = new HashMap<>();
        
        for (UserBehavior behavior : behaviors) {
            // 确定宠物类型
            String petType = extractPetType(behavior);
            if (petType == null) continue;
            
            // 提取年龄阶段关键词
            Set<String> ageStages = extractAgeStages(behavior);
            
            // 计算时间衰减
            double timeDecay = calculateTimeDecay(behavior.getTimestamp());
            
            // 累加分数
            for (String ageStage : ageStages) {
                double baseWeight = getBaseWeight(behavior, ageStage);
                double finalWeight = baseWeight * timeDecay;
                
                petAgeScores
                    .computeIfAbsent(petType, k -> new HashMap<>())
                    .merge(ageStage, finalWeight, Double::sum);
            }
        }
        
        // 生成年龄标签
        for (Map.Entry<String, Map<String, Double>> entry : petAgeScores.entrySet()) {
            PetAgeLabel label = generateAgeLabel(userId, entry.getKey(), entry.getValue());
            out.collect(label);
        }
    }
    
    private String extractPetType(UserBehavior behavior) {
        // 简化实现，从类目推断宠物类型
        if (behavior.getCategory1().contains("狗")) return "狗";
        if (behavior.getCategory1().contains("猫")) return "猫";
        if (behavior.getCategory1().contains("仓鼠")) return "仓鼠";
        if (behavior.getCategory1().contains("鸟")) return "鸟类";
        if (behavior.getCategory1().contains("水族")) return "水族";
        return null;
    }
    
    private Set<String> extractAgeStages(UserBehavior behavior) {
        Set<String> stages = new HashSet<>();
        
        // 从类目属性提取
        Map<String, String> attributes = behavior.getItemAttributes();
        if (attributes != null) {
            String ageAttr = attributes.get("age_stage");
            if (ageAttr != null) {
                stages.addAll(Arrays.asList(ageAttr.split(",")));
            }
        }
        
        // 从标题和描述中提取关键词
        String title = attributes != null ? attributes.get("title") : "";
        
        stages.addAll(extractKeywords(title));
        
        return stages;
    }

    private Set<String> extractKeywords(String text) {
        Set<String> keywords = new HashSet<>();
        if (text == null || text.isEmpty()) return keywords;

        String lowerText = text.toLowerCase();

        // 幼年阶段关键词
        if (lowerText.contains("幼犬") || lowerText.contains("幼猫") ||
                lowerText.contains("奶狗") || lowerText.contains("奶猫") ||
                lowerText.contains("puppy") || lowerText.contains("kitten") ||
                lowerText.contains("幼年")) {
            keywords.add("幼年");
        }
        // 成年阶段关键词
        if (lowerText.contains("成犬") || lowerText.contains("成猫") ||
                lowerText.contains("成年") || lowerText.contains("adult")) {
            keywords.add("成年");
        }
        // 老年阶段关键词
        if (lowerText.contains("老年") || lowerText.contains("老犬") ||
                lowerText.contains("老猫") || lowerText.contains("senior")) {
            keywords.add("老年");
        }

        return keywords;
    }
    
    private double calculateTimeDecay(Long timestamp) {
        long daysDiff = (System.currentTimeMillis() - timestamp) / (1000 * 60 * 60 * 24);
        int monthDiff = (int) (daysDiff / 30);
        return Math.pow(0.9, monthDiff); // 每30天衰减0.9
    }
    
    private double getBaseWeight(UserBehavior behavior, String ageStage) {
        Map<String, Double> weights = new HashMap<>();
        weights.put("buy", 5.0);
        weights.put("cart", 3.0);
        weights.put("favorite", 2.0);
        weights.put("browse", 1.0);
        
        return weights.getOrDefault(behavior.getBehaviorType(), 0.0);
    }
    
    private PetAgeLabel generateAgeLabel(String userId, String petType, 
                                       Map<String, Double> ageScores) {
        PetAgeLabel label = new PetAgeLabel();
        label.setUserId(userId);
        label.setPetType(petType);
        label.setAgeScores(ageScores);
        
        // 应用阈值规则
        List<String> finalLabels = new ArrayList<>();
        
        // 强相关标签
        ageScores.entrySet().stream()
            .filter(entry -> entry.getValue() >= 20.0)
            .forEach(entry -> finalLabels.add(entry.getKey()));
        
        // 弱相关标签处理
        if (finalLabels.isEmpty()) {
            ageScores.entrySet().stream()
                .filter(entry -> entry.getValue() >= 10.0)
                .sorted(Map.Entry.<String, Double>comparingByValue().reversed())
                .limit(2)
                .forEach(entry -> finalLabels.add(entry.getKey()));
        }
        
        label.setAgeLabels(finalLabels);
        label.setUpdateTime(System.currentTimeMillis());
        
        return label;
    }
}